package com.dongdongshop.MQ;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.dongdongshop.mapper.TbSeckillGoodsMapper;
import com.dongdongshop.mapper.TbSeckillOrderMapper;
import com.dongdongshop.pojo.TbSeckillGoods;
import com.dongdongshop.pojo.TbSeckillOrder;
import com.dongdongshop.util.IdWorker;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;


@Component
@RocketMQMessageListener( topic = "seckill_topic", consumerGroup = "seckill_consum")
public class SeckillConsumer implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {


    @Autowired
    StringRedisTemplate stringRedisTemplate;
    @Autowired
    IdWorker idWorker;
    @Autowired
    private TbSeckillGoodsMapper tgm;//秒杀商品表
    @Autowired
    private TbSeckillOrderMapper tom;//秒杀订单表
    @Autowired
    RedisTemplate redisTemplate;




    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //接收我们mq的消息
                MessageExt messageExt=list.get(0);
                int reconsumeTimes = messageExt.getReconsumeTimes();//重试次数
                System.out.println("重试次数为"+reconsumeTimes);
                try {
                    messageExt.getTopic();//接收topic
                    messageExt.getMsgId();//消息id
                    //判断传来的消息是否存在，redis中建一个id表
                    String msgId = messageExt.getMsgId();
                    Boolean hasKey = stringRedisTemplate.hasKey("MESSAGE" + msgId);
                    if(!hasKey){
                        byte[] body = messageExt.getBody();//消息内容
                        String msg= new String (body);
                        System.out.println("消息内容为"+msg);
                        //字符串类型json转json类
                        JSONObject json = JSONUtil.parseObj(msg);
                        Object username = json.get("username");//用户名
                        Object sekillid = json.get("sekillid");//商品id
                        Long aLong = Long.valueOf(sekillid.toString());
                        TbSeckillGoods goods = tgm.selectByPrimaryKey(aLong);
                        //添加订单信息
                        TbSeckillOrder order=new TbSeckillOrder();
                        long l = idWorker.nextId();
                        order.setId(l);//订单id
                        order.setStatus("0");//未支付
                        order.setUserId(username.toString());//用户名
                        order.setCreateTime(new Date());//创建时间
                        order.setSellerId(goods.getSellerId());
                        order.setMoney(goods.getCostPrice());//要支付的钱
                        order.setSeckillId(aLong);
                        tom.insertSelective(order);
                        //下单成功后，将订单id放入redis缓存中
                        redisTemplate.boundHashOps("seckill_orderid").put(username.toString(),l);

//                        int x=9/0;
                        stringRedisTemplate.boundValueOps("MESSAGE"+msgId).set("已消费");
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//消费成功返回

                }catch (Exception e){
                    e.printStackTrace();
                    if(reconsumeTimes<2){//如果重试次数小于2继续重试，大于2就消费成功，放入死信队列
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;//消费失败重试
                    }
                    System.out.println("消息重试次数>2次，加入死信队列，等待人工补偿");
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//消费成功返回
                }

            }
        });
    }

    @Override
    public void onMessage(String s) {

    }
}
